package rabbitmq.test;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 专门检测生产发送的逻辑
 */
public class MyProducer {
    //准备连接通道
    //每一个测试类都准备一个连接对象
    private Channel channel;

    @Before
    public void init() {
        //1.创建长连接 提供属性(ip,port,username,password,virtualHost)
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.8.100");
        factory.setPort(5672);//TCP协议端口
        factory.setUsername("admin");
        factory.setPassword("123456");
        factory.setVirtualHost("/");
        try {
            Connection connection = factory.newConnection();
            //2.从长连接赋值短链接
            channel = connection.createChannel();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("创建连接失败");
        }
    }

    //发送简单的消息
    @Test
    public void sendSimpleMessage() throws Exception {
        //调用channel的api发送消息
        String message = "你好rabbitmq,hello world";
        /**
         * String exchange: 生产者必须将消息发送到交换机,不能直接发到队列,指定交换机名称 使用的是默认交换机
         * String routingKey: 路由键,指定交换机将消息发送到哪个队列,使用默认交换机,路由键就是队列名称
         * BasicProperties props: 消息的属性,比如消息的优先级,是否持久化,消息的存活时间等
         * 优先级逻辑: 1000条消息 优先级都是1 有一个优先级是100 消费者接到1000个消息 先处理优先级高的
         * byte[] body: 消息体,就是发送的消息,如果发送内存数据 需要先进行序列化处理
         */
        channel.basicPublish(
                "",
                "queue01",
                null,
                message.getBytes(StandardCharsets.UTF_8));
        System.out.println("消息发送成功");
    }

    @Test
    public void sendTxMessage() throws IOException {
        String message = "你好rabbitmq,hello world";
        try {
            //开启同步事务发送消息
            channel.txSelect();
            channel.basicPublish("", "queue01", null, message.getBytes(StandardCharsets.UTF_8));
            //线程阻塞等待发送结果
            AMQP.Tx.CommitOk commitOk = channel.txCommit();
            System.out.println("本次发送事务结束,结果:{}" + commitOk);
        } catch (Exception e) {
            //如果出现发送异常情况,比如连接超时,等待超时 关闭事务回滚
            channel.txRollback();
            System.out.println("消息发送失败,已回滚");
        }
    }

    //等待确认发布
    @Test
    public void sendConfirmMessage() throws IOException, InterruptedException, TimeoutException {
        //开启确认模式
        channel.confirmSelect();
        //正常发送消息
        String message = "你好rabbitmq,hello world";
        channel.basicPublish("", "queue01", null, message.getBytes(StandardCharsets.UTF_8));
        //设置超时等待时间(一旦超过时间上线 判断失败)
        boolean result = channel.waitForConfirms(500);//超过500毫秒就失败
        if (result) {
            System.out.println("消息发送成功");
        } else {
            System.out.println("消息发送失败");
        }
    }

    // 发送消息到自定义的路由交换机
    @Test
    public void sendDirectMessage() throws IOException {
        String msg = "路由交换机 消息";
        channel.basicPublish("direct_ex", "成都", null, msg.getBytes());
    }

    // 发送消息到自定义的发布订阅交换机
    @Test
    public void sendFanoutMessage() throws IOException {
        String msg = "发布订阅交换机 消息";
        channel.basicPublish("fanout_ex", "北京", null, msg.getBytes());
    }

    // 发送消息到自定义的主题交换机
    @Test
    public void sendTopicMessage() throws IOException {
        String msg = "主题交换机 消息";
        channel.basicPublish("topic_ex", "中国.石家庄", null, msg.getBytes());
    }

    // 发送消息到头交换机
    @Test
    public void sendHeadersMessage() throws IOException {
        String msg = "头交换机 消息";
        // 携带消息的属性头
        // 创建消息头（注意键名需要是String类型）
        Map<String, Object> headers = new HashMap<>();
        headers.put("format", "pdf");    // 示例头信息1
        headers.put("type", "report");   // 示例头信息2

        // 创建AMQP属性对象并封装头信息
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .headers(headers)
                .build();
        channel.basicPublish("", "queue01", props, msg.getBytes());  // 默认绑定，不要写交换机的路由key
    }

    // 发送消息到混合交换机
    @Test
    public void sendMixMessage() throws IOException {
        String msg = "混合交换机-向路由 消息";
        channel.basicPublish("direct_ex2", "重庆", null, msg.getBytes());
    }
}
